package com.gd.data.platform.service.realtime;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.gd.data.platform.config.RealtimeDataWebSocketHandler;
import com.gd.data.platform.dto.RealtimeMonitorData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

/**
 * 实时数据推送服务
 * 负责向前端推送实时监控数据
 * 
 * @author gd-data-platform
 * @since 1.0.0
 */
@Slf4j
@Service
public class RealtimeDataPushService {

    @Autowired
    private RealtimeMonitorService realtimeMonitorService;

    @Autowired
    private RealtimeDataWebSocketHandler webSocketHandler;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private ObjectMapper objectMapper;

    private static final String REALTIME_DATA_TOPIC = "realtime-monitor-data";

    /**
     * 定时推送实时数据（每10秒）
     */
    @Scheduled(fixedRate = 10000)
    @Async
    public void pushRealtimeData() {
        try {
            // 获取实时监控数据
            RealtimeMonitorData data = realtimeMonitorService.getRealtimeMonitorData();

            // 检查告警条件
            realtimeMonitorService.checkAlertConditions(data);

            // 转换为JSON
            String jsonData = objectMapper.writeValueAsString(data);

            // 通过WebSocket推送给前端
            webSocketHandler.sendRealtimeData(jsonData);

            // 发送到Kafka（用于其他系统消费）
            kafkaTemplate.send(REALTIME_DATA_TOPIC, jsonData);

            log.debug("实时数据推送完成，连接数：{}", webSocketHandler.getConnectionCount());

        } catch (Exception e) {
            log.error("推送实时数据失败", e);
        }
    }

    /**
     * 监听Kafka消息（API调用日志）
     */
    @KafkaListener(topics = "api-call-logs", groupId = "realtime-monitor")
    public void handleApiCallLog(String message) {
        try {
            log.debug("收到API调用日志：{}", message);

            // 这里可以实时处理API调用日志
            // 例如：实时更新缓存、触发告警等

        } catch (Exception e) {
            log.error("处理API调用日志失败", e);
        }
    }

    /**
     * 监听Kafka消息（下载日志）
     */
    @KafkaListener(topics = "download-logs", groupId = "realtime-monitor")
    public void handleDownloadLog(String message) {
        try {
            log.debug("收到下载日志：{}", message);

            // 这里可以实时处理下载日志
            // 例如：实时更新下载统计、触发告警等

        } catch (Exception e) {
            log.error("处理下载日志失败", e);
        }
    }

    /**
     * 手动推送实时数据
     */
    public void pushRealtimeDataManually() {
        pushRealtimeData();
    }
}
